package com.google.code.stalkernet.reactor;

import java.io.IOException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.ArrayDeque;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * An implementation of {@link Reactor} interface based on {@link Selector}.
 *
 * @author Mikhail Vladimirov
 */
public class SelectorReactor implements Reactor, Runnable
{
    private final static Logger logger =
        Logger.getLogger (SelectorReactor.class.getName ());

    private final Selector selector;
    private final Queue <TimedRunnable> timedWorkQueue =
        new PriorityQueue <> ();
    private final Queue <Runnable> workQueue = new ArrayDeque <> ();
    private final Queue <SelectionKey> selectionKeyQueue =
        new ArrayDeque <> ();

    boolean shutdown = false;

    /**
     * Create new selector reactor that uses given {@link Selector}.
     *
     * @param selector {@link Selector} to use
     */
    public SelectorReactor (Selector selector)
    {
        if (selector == null)
            throw new IllegalArgumentException ("Selector is null");

        this.selector = selector;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public SelectionKey registerSelectableChannel (
        SelectableChannel selectableChannel,
        SelectableChannelCallback selectableChannelCallback)
        throws IOException
    {
        if (selectableChannel == null)
            throw new IllegalArgumentException ("Selectable channel is null");

        if (selectableChannelCallback == null)
            throw new IllegalArgumentException (
                "Selectable channel callback");

        return selectableChannel.register (
            selector, 0, selectableChannelCallback);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public ScheduledTaskHandle schedule (Runnable runnable, long time)
    {
        if (runnable == null)
            throw new IllegalArgumentException ("Runnable is null");

        TimedRunnable timedRunnable = new TimedRunnable (runnable, time);

        timedWorkQueue.add (timedRunnable);

        return timedRunnable;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void invoke (Runnable runnable)
    {
        if (runnable == null)
            throw new IllegalArgumentException ("Runnable is null");

        synchronized (workQueue)
        {
            workQueue.add (runnable);
        }

        selector.wakeup ();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void shutdown ()
    {
        shutdown = true;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void run ()
    {
        try
        {
            while (!shutdown)
            {
                Runnable runnable = getNextRunnable ();

                if (runnable != null)
                {
                    try
                    {
                        runnable.run ();
                    }
                    catch (Throwable ex)
                    {
                        logger.log (
                            Level.SEVERE,
                            "Exception in runnable: " + runnable,
                            ex);
                    }
                }
                else doSelect ();
            }
        }
        catch (IOException ex)
        {
            logger.log (Level.SEVERE, "Cannot select", ex);
        }
    }

    private Runnable getNextRunnable ()
    {
        Runnable runnable = null;

        if (runnable == null) runnable = getNextWorkRunnable ();
        if (runnable == null) runnable = getNextTimedWorkRunnable ();
        if (runnable == null) runnable = getNextSelectionRunnable ();

        return runnable;
    }

    private Runnable getNextWorkRunnable ()
    {
        synchronized (workQueue)
        {
            return workQueue.poll ();
        }
    }

    private Runnable getNextTimedWorkRunnable ()
    {
        TimedRunnable timedRunnable = timedWorkQueue.peek ();

        if (timedRunnable != null &&
            timedRunnable.getTime () <= System.currentTimeMillis ())
            return timedWorkQueue.poll ();
        else return null;
    }

    private Runnable getNextSelectionRunnable ()
    {
        SelectionKey selectionKey = selectionKeyQueue.poll ();

        return selectionKey != null ?
            new SelectionKeyRunnable (selectionKey) : null;
    }

    private void doSelect () throws IOException
    {
        TimedRunnable timedRunnable = timedWorkQueue.peek ();

        if (timedRunnable == null) selector.select ();
        else
        {
            long timeout =
                timedRunnable.getTime () - System.currentTimeMillis ();

            if (timeout > 0L) selector.select (timeout);
            else selector.selectNow ();
        }

        selectionKeyQueue.addAll (selector.selectedKeys ());
    }

    private class TimedRunnable
        implements Runnable, Comparable <TimedRunnable>, ScheduledTaskHandle
    {
        private final Runnable runnable;
        private final long time;

        public TimedRunnable (Runnable runnable, long time)
        {
            if (runnable == null)
                throw new IllegalArgumentException ("Runnable is null");

            this.runnable = runnable;
            this.time = time;
        }

        public long getTime ()
        {
            return time;
        }

        @Override
        public int compareTo (TimedRunnable o)
        {
            if (o == null)
                throw new IllegalArgumentException ("O is null");

            return Long.compare (time, o.time);
        }

        @Override
        public void run ()
        {
            runnable.run ();
        }

        @Override
        public boolean cancel ()
        {
            return timedWorkQueue.remove (this);
        }
    }

    private static class SelectionKeyRunnable implements Runnable
    {
        private final SelectionKey selectionKey;

        public SelectionKeyRunnable (SelectionKey selectionKey)
        {
            if (selectionKey == null)
                throw new IllegalArgumentException ("Selection key is null");

            this.selectionKey = selectionKey;
        }

        @Override
        public void run ()
        {
            ((SelectableChannelCallback)selectionKey.attachment ()).
                selected (selectionKey);
        }
    }
}
